package com.zhengb.flink.study.datastream.transformation;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.datastream.datasource.GoodsSellDataSource;
import com.zhengb.flink.study.datastream.entity.GoodSell;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * Split
 * DataStream -> SplitStream
 *
 * @author zhengbo
 * @date 2019/12/20
 */
@Slf4j
public class ReduceTransformation {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new GoodsSellDataSource()).keyBy(new KeySelector<GoodSell, String>() {
            //按商品名称分发
            @Override
            public String getKey(GoodSell value) throws Exception {
                return value.getGoodName();
            }
        }).reduce(new ReduceFunction<GoodSell>() {
            @Override
            public GoodSell reduce(GoodSell value1, GoodSell value2) throws Exception {
                //将两个商品销量 销售额规约为一个
                log.info("reduce,selle1:{},sell2:{}", JSONObject.toJSONString(value1), JSONObject.toJSONString(value2));

                return new GoodSell(value1.getGoodName(), value1.getSellNum() + value2.getSellNum(),
                        value1.getSellAmount() + value2.getSellAmount());
            }
        }).addSink(new RichSinkFunction<GoodSell>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }

            //打印商品当前销量 销售总额
            @Override
            public void invoke(GoodSell value, Context context) throws Exception {
                log.info("sink data:{}", JSONObject.toJSONString(value));
            }
        });

        env.execute("ReduceTransformation");

    }
}
